package org.atguigu.gmall.realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.atguigu.gmall.realtime.app.BaseSQLApp;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.util.SQLUtil;

import java.time.Duration;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dwd.db
 * Author: ZengHaiFeng
 * CreateTime:2023/4/26 10:38:08
 * Description: Todo
 * Version：1.0
 */
public class Dwd_07_DwdTradeOrderRefund extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_07_DwdTradeOrderRefund().init(30007,2,"Dwd_07_DwdTradeOrderRefund");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tbe) {
        tbe.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        // read data from ods_db
        readOdsDb(tbe,"Dwd_07_DwdTradeOrderRefund");
        // read data from base_dic
        readBaseDic(tbe);
        // filter order refund table data
        Table orderRefundInfo = tbe.sqlQuery(" select" +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " `data`['order_id'] order_id, " +
                " `data`['sku_id'] sku_id, " +
                " `data`['refund_type'] refund_type, " +
                " `data`['refund_num'] refund_num, " +
                " `data`['refund_amount'] refund_amount, " +
                " `data`['refund_reason_type'] refund_reason_type, " +
                " `data`['refund_reason_txt'] refund_reason_txt, " +
                " `data`['create_time'] create_time, " +
                " `pt`, " +
                " `ts`  " +
                " from ods_db where " +
                " `database`='gmall2023' and " +
                " `table`='order_refund_info' and " +
                " `type`='insert'");
        tbe.createTemporaryView("order_refund_info",orderRefundInfo);

        // filter order info table data that just need province dada
        Table orderInfo = tbe.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['province_id'] province_id, " +
                " `old`  " +
                "from ods_db where " +
                " `database`='gmall2023' and " +
                " `table`='order_info' and " +
                " `type`='update' and " +
                " `old`['order_status'] is not null and " +
                " `data`['order_status']='1005' ");
        tbe.createTemporaryView("order_info",orderInfo);

        // then there are three tables join
        Table result = tbe.sqlQuery("select " +
                " `or`.id," +
                " `or`.user_id, " +
                " `or`.order_id, " +
                " `or`.sku_id, " +
                " `oi`.province_id," +
                " date_format(`or`.create_time,'yyyy-MM-dd') date_id," +
                " `or`.create_time," +
                " `or`.refund_type refund_type_code," +
                " `dic1`.dic_name refund_type_name," +
                " `or`.refund_reason_type refund_reason_code," +
                " `dic2`.dic_name refund_reason_name," +
                " `or`.refund_reason_txt," +
                " `or`.refund_num," +
                " `or`.refund_amount," +
                " `or`.ts  " +
                " from order_refund_info as `or` join order_info as `oi` " +
                " on `or`.order_id=`oi`.id " +
                " join `base_dic` for system_time as of `or`.pt as `dic1` " +
                " on `or`.refund_type=`dic1`.dic_code " +
                " join `base_dic` for system_time as of `or`.pt as `dic2` " +
                " on `or`.refund_reason_type=`dic2`.dic_code ");

        // finally output result into kafka topic
        tbe.executeSql("create table dwd_trade_order_refund(" +
                " id string," +
                " user_id string," +
                " order_id string, " +
                " sku_id string, " +
                " province_id string, " +
                " date_id string, " +
                " create_time string, " +
                " refund_type_code string, " +
                " refund_type_name string, " +
                " refund_reason_code string, " +
                " refund_reason_name string, " +
                " refund_reason_txt string, " +
                " refund_num string, " +
                " refund_amount string, " +
                " ts bigint " +
                ")" +
                SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_ORDER_REFUND));
        result.executeInsert("dwd_trade_order_refund");
    }
}
